package com.atguigu.actual.edu0417.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.actual.edu0417.common.EduConfig;
import com.atguigu.actual.edu0417.unils.HbaseUtil;
import com.atguigu.actual.edu0417.unils.RedisUtil;
import com.google.common.util.concurrent.AsyncFunction;
import com.google.common.util.concurrent.ListenableFuture;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.async.RedisAsyncCommands;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.hadoop.hbase.client.AsyncConnection;
import org.checkerframework.checker.nullness.qual.Nullable;

import java.util.Collections;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * @author: techyang
 * @since: 2023-10-22 08:28
 * @description: 异步获取维度函数
 **/
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T,T> implements DimJoinFunction<T> {

    // redis异步连接
    private StatefulRedisConnection<String,String> redisConnection;
    // habse异步连接
    private AsyncConnection asyncConnection;

    // 表名
    private String tableName;

    public DimAsyncFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        redisConnection = RedisUtil.getAsyncRedisConnection();
        asyncConnection = HbaseUtil.getAsyncConnection();
    }

    @Override
    public void close() throws Exception {
        RedisUtil.closeAysncRedisConnection(redisConnection);
        HbaseUtil.closeAsyncConnection(asyncConnection);
    }

    @Override
    public void asyncInvoke(T bean, ResultFuture<T> resultFuture) throws Exception {
        // 从Redis中获取维度信息
        CompletableFuture<Void> completableFuture =
                CompletableFuture.supplyAsync(
                        new Supplier<JSONObject>() {
                            @Override
                            public JSONObject get() {
                                JSONObject dimInfoJsonObj = RedisUtil.asyncGetDimInfo(redisConnection, tableName + ":" + getKey(bean));
                                return dimInfoJsonObj;
                            }
                        }
                ).thenApplyAsync(new Function<JSONObject, JSONObject>() {
                    @Override
                    public JSONObject apply(JSONObject dimInfoJsonObj) {
                        if (Objects.isNull(dimInfoJsonObj)) {
                            // redis中没有获取到维度信息
                            dimInfoJsonObj = HbaseUtil.getDimInfoFromHbaseByAsync(asyncConnection, EduConfig.HBASE_NAMESPACE, tableName, getKey(bean));
                            if (Objects.nonNull(dimInfoJsonObj)) {
                                System.out.println(" 从HBase中异步获取到 <" + tableName + ":" + getKey(bean) + "> 数据");
                                // 将HBase中的数据异步写入Redis
                                RedisUtil.asyncWriteDim(redisConnection,tableName + ":" + getKey(bean),dimInfoJsonObj);
                                return dimInfoJsonObj;
                            } else {
                                System.out.println(" 从HBase中<没有>异步获取到 <" + tableName + ":" + getKey(bean) + "> 数据");
                            }
                        } else {
                            System.out.println(" 从Redis中异步获取到 <" + tableName + ":" + getKey(bean) + "> 数据");
                            return dimInfoJsonObj;
                        }
                        return dimInfoJsonObj;
                    }
                }).thenAcceptAsync(new Consumer<JSONObject>() {
                    @Override
                    public void accept(JSONObject dimInfoJsonObj) {
                        if (Objects.nonNull(dimInfoJsonObj)) {
                            join(bean,dimInfoJsonObj);
                        }
                        resultFuture.complete(Collections.singleton(bean));
                    }
                });
    }

    // 如果异步超时,会执行这个方法
    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        // 根据自己情况进行处理
        throw new RuntimeException("异步超时:一般是其他原因导致的异步超时, 请检查: \n" +
                "1.检查集群是否都 ok: hdfs redis hbase kafka. \n" +
                "2.检查下 redis 的配置 bind 0.0.0.0 \n" +
                "3.检查用到的6张维度表是否都在,并且每张表都数据. 最好通过 maxwell-bootstrap 同步一下 \n" +
                "4. 找我" +
                "");
    }
}
